package sunyu.service.hbase.impl;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import sunyu.service.hbase.HbaseService;
import sunyu.tools.common.TimeTools;
import sunyu.tools.hbase.HbaseTools;
import sunyu.tools.hbase.pojo.Row;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

@Service("hbaseService")
public class HbaseServiceImpl implements HbaseService {

    private static final Logger logger = LoggerFactory.getLogger(Thread.currentThread().getClass());

    @Resource(name = "hbaseTools")
    private HbaseTools hbaseTools;

    @Resource(name = "timeTools")
    private TimeTools timeTools;

    @Override
    public List<Row> find(String tableName,
                          String did,
                          String startTime,
                          String endTime,
                          Boolean reversed,
                          FilterList filterList,
                          Integer pageSize,
                          Boolean isFirstPage,
                          String startRowName,
                          List<String> familys,
                          List<String> columns) {

        if (reversed == null) {
            reversed = false;
        }
        if (pageSize == null) {
            pageSize = 10;
        }

        String pre, startRow, stopRow, a, b;
        List<Row> result;
        if (filterList == null) {
            filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);// FilterList.Operator.MUST_PASS_ALL -->and ，FilterList.Operator.MUST_PASS_ONE -->or
        }
        if (StringUtils.isBlank(startTime)) {// 如果开始时间为空，则去查询redis或者mysql，将当前设备的创建时间赋予这个变量
            startTime = "20000101000000";
        }
        if (StringUtils.isBlank(endTime)) {// 如果结束时间为空，则去查询redis或者mysql，将设备最后定位时间赋予这个变量
            endTime = timeTools.toString(new Date(), "yyyyMMddHHmmss");
        }

        pre = did + "_";// 设备+分隔符(hbase的rowKey必须是did_日期时间格式)

        Filter preFilter = new PrefixFilter(Bytes.toBytes(pre));
        filterList.addFilter(preFilter);

        startRow = pre + startTime;
        stopRow = pre + endTime;

        if (reversed) {// 倒序
            a = stopRow;
            b = startRow;
        } else {
            a = startRow;
            b = stopRow;
        }
        b += "0";// 加0的意义是返回本条数据

        if (StringUtils.isNotBlank(startRowName)) {
            a = startRowName;
        }

        logger.debug("表名：" + tableName);
        logger.debug("did：" + did);
        logger.debug("每页显示记录数：" + pageSize);
        logger.debug("startRow：" + a);
        logger.debug("stopRow：" + b);
        logger.debug("是否倒序：" + reversed);

        result = hbaseTools.find(tableName, pageSize, isFirstPage, a, b, reversed, filterList, familys, columns);
        return result;
    }

    @Override
    public Long count(String tableName,
                      String did,
                      String startTime,
                      String endTime,
                      FilterList filterList) {
        if (StringUtils.isBlank(did)) {//没有did就是统计所有了，太慢了。。。
            return hbaseTools.count(tableName, filterList);
        }

        String pre, startRow, stopRow, a, b;
        if (filterList == null) {
            filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);// FilterList.Operator.MUST_PASS_ALL -->and ，FilterList.Operator.MUST_PASS_ONE -->or
        }
        if (StringUtils.isBlank(startTime)) {// 如果开始时间为空，则去查询redis或者mysql，将当前设备的创建时间赋予这个变量
            startTime = "20000101000000";
        }
        if (StringUtils.isBlank(endTime)) {// 如果结束时间为空，则去查询redis或者mysql，将设备最后定位时间赋予这个变量
            endTime = timeTools.toString(new Date(), "yyyyMMddHHmmss");
        }

        pre = did + "_";// 设备+分隔符(hbase的rowKey必须是did_日期时间格式)

        Filter preFilter = new PrefixFilter(Bytes.toBytes(pre));
        filterList.addFilter(preFilter);

        startRow = pre + startTime;
        stopRow = pre + endTime;

        a = startRow;
        b = stopRow;
        b += "0";// 加0的意义是返回本条数据

        logger.debug("表名：" + tableName);
        logger.debug("did：" + did);
        logger.debug("startRow：" + a);
        logger.debug("stopRow：" + b);

        return hbaseTools.count(tableName, filterList, startRow, stopRow);
    }

}
